package innermq

import (
	"context"
	"time"

	"iot-base/common/common"
	"iot-base/common/config"
	"iot-base/common/logger"
	rdbc "iot-base/common/redis"

	"github.com/go-redis/redis/v8"

	"github.com/vmihailenco/msgpack/v5"
)

var pubmsg = ".pubmsg"

var msglist = "p2p.msg.list"

//MSGFUN 消息处理函数
type MSGFUN func(topic string, body interface{})

//MSG  消息
type MSG struct {
	Body interface{}
}

var rc *redis.Client
var ctx = context.Background()

func init() {
	pubmsg = pubmsg + "@" + config.PType
	rc = rdbc.GetRCClient()
	ctx = rdbc.GetRCContext()
}

//P2PMsgPub 发布p2p消息
func P2PMsgPub(topic string, obj interface{}) {
	/*msg := &MSG{
		Body: obj,
	}*/

	bmsg, err := msgpack.Marshal(obj)
	if err == nil {
		//fmt.Println("send msg--------------------------")
		rc.RPush(ctx, topic+msglist, bmsg)
	}

}

type TIMEOUTFUN func(topic string)

func unPackMsg(body interface{}, data []byte) interface{} {
	//var msgObj MSG
	//var err error
	var currError error
	if body != nil {
		obody, err := common.CreateInstance(body)
		if err == nil {

			err = msgpack.Unmarshal(data, obody)
			if err == nil {
				return obody
			}
			currError = err

			//msgObj.Body = make(map[string]interface{})
		}
		currError = err
		logger.SLog.Error("cant anto change body to instance", currError)
	} else {
		logger.SLog.Error("unpack msg send a body is null")
	}

	return data
	//return data
}
func WaitforMsg(topic string, body interface{}, timeout time.Duration, fun MSGFUN, funtimeout TIMEOUTFUN) {
	//只有p2p消息才能进行wait
	//fmt.Println("waitformsg------------*************-", topic+msglist)
	waitformsgImpl(topic, body, timeout, fun, funtimeout, false)
}

func WaitforMsgSync(topic string, body interface{}, timeout time.Duration, fun MSGFUN, funtimeout TIMEOUTFUN) {
	//只有p2p消息才能进行wait
	//fmt.Println("waitformsg------------*************-", topic+msglist)
	waitformsgImpl(topic, body, timeout, fun, funtimeout, true)
}

func waitformsgImpl(topic string, body interface{}, timeout time.Duration, fun MSGFUN, funtimeout TIMEOUTFUN, bwait bool) {
	//只有p2p消息才能进行wait
	//fmt.Println("waitformsg------------*************-", topic+msglist)
	//var bctx = context.Background()

	//rc.ClientSetName(ctx, topic)
	msgs, err := rc.BLPop(ctx, timeout, topic+msglist).Result()
	//fmt.Println("wait msg---------------wakeup-----------", len(msgs))

	if (err == nil) && len(msgs) == 2 {
		//msg := msgs[0]
		logger.SLog.Error("******sucess waitformsgImpl:", topic)
		msgObj := unPackMsg(body, []byte(msgs[1]))
		if !bwait {
			go fun(topic, msgObj)
		} else {
			fun(topic, msgObj)
		}

	} else {
		logger.SLog.Error("******error waitformsgImpl:", err, len(msgs), topic)
		if funtimeout != nil {
			if bwait {
				funtimeout(topic)
			} else {
				go funtimeout(topic)
			}

		}
	}
}

func p2PMsgSub(topic string, body interface{}, fun MSGFUN) {
	var bctx = context.Background()
	brc := rc.Conn(bctx)
	for {
		//fmt.Println("wait msg--------------------------")
		//WaitforMsg(topic, body, 0, fun, nil)
		//断开后重新连接

		_, errC := brc.ClientSetName(bctx, topic).Result()
		if errC == nil {
			//logger.SLog.Info("long connection ------begin-BLPop---", topic)

			msgs, err := brc.BLPop(bctx, 0, topic+msglist).Result()
			if (err == nil) && len(msgs) == 2 {
				//msg := msgs[0]
				//logger.SLog.Info("***long***sucess BLPop:", topic)
				msgObj := unPackMsg(body, []byte(msgs[1]))
				fun(topic, msgObj)
				continue

			} else {
				logger.SLog.Error("long connection --BLPop error----", topic, err)
			}
		}
		//logger.SLog.Error("long connection --set clinet error-reconnect---", topic, errC)
		for {

			brc.Close()
			brc = rc.Conn(bctx)
			status, perr := brc.Ping(bctx).Result()
			if perr == nil {
				logger.SLog.Info("long connection reconnect--ok-", topic, errC, status)
				break
			}
			time.Sleep(time.Second * 10)
			logger.SLog.Error("long connection-reconnect---", topic, perr, status)
		}

	}

}

//SubP2PMsg 订阅P2P消息
func P2PMsgSub(topic string, body interface{}, fun MSGFUN) {
	go p2PMsgSub(topic, body, fun)

}

//MsgPub 发布广播消息
func MsgPub(topic string, obj interface{}) {
	var rcerr error
	//fmt.Println("msg----------------------send--------")
	if obj != nil {
		bmsg, err := msgpack.Marshal(obj)
		if err == nil {
			rcerr = rc.Publish(ctx, topic+pubmsg, bmsg).Err()

		} else {
			logger.SLog.Error("msg pub msgpack error", err)
		}
	} else {
		rcerr = rc.Publish(ctx, topic+pubmsg, nil).Err()
	}
	if rcerr != nil {
		logger.SLog.Error("publish has error", rcerr)
	}

}
func msgSub(topic string, body interface{}, isAlltime bool, fun MSGFUN) {
	logger.SLog.Info("*********msgSub********", topic, isAlltime)
	handler := rc.Subscribe(ctx, topic+pubmsg)
	defer handler.Close()
	_, err := handler.Receive(ctx)
	if err != nil {
		logger.SLog.Error("msg sub error receive error", err)
		return
	}
	/*for {
		obj, err := handler.ReceiveMessage(ctx)
		//fmt.Println("recv-------------", reflect.TypeOf(obj), obj)
		if err == nil {

			msg := unPackMsg(body, []byte(obj.Payload))
			go fun(topic, msg)

		}
		if !isAlltime {
			break
		}
	}*/
	//Recieve 存在隐患，改用channel方式
	ch := handler.Channel()
	for obj := range ch {
		msg := unPackMsg(body, []byte(obj.Payload))
		go fun(topic, msg)
		if !isAlltime {
			break
		}
	}
}

//MsgSub 订阅广播消息
func MsgSub(topic string, body interface{}, fun MSGFUN) {
	go msgSub(topic, body, true, fun)

}
